package com.lyh.dt.utils;

import com.lyh.dt.server.DtActiveMqBean;
import com.lyh.dt.server.TransactionClient;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.io.Serializable;

/**
 * @author liuyuhong
 */
public class DtActiveMqUtil {

    private static final Logger logger = LoggerFactory.getLogger(DtActiveMqUtil.class);

    private static Connection connection;

    public static final String ACTIVEMQ_URL = "tcp://192.168.1.220:61616";


    private static final String QUEUE_TO_TM = "queueToTm";

    private static final String TOPIC_TO_RM = "topicToRm";

    static {
        //创建连接工厂，按照给定的url地址，采用默认用户名和密码（admin）
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        activeMQConnectionFactory.setTrustAllPackages(true);
        //通过连接工厂，获得连接connection并启动
        try {
            connection = activeMQConnectionFactory.createConnection();
            connection.start();
        } catch (JMSException e) {
            logger.error("获取activeMQ连接失败", e);
        }
    }

    /**
     * 向TM发送消息
     * @param obj
     */
    public static void sendDtMessageToTm(DtActiveMqBean obj){
        try {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(QUEUE_TO_TM);
            MessageProducer producer = session.createProducer(queue);
            ObjectMessage objectMessage = session.createObjectMessage(obj);
            producer.send(objectMessage);
        } catch (JMSException e) {
            logger.error("向TM发送消息失败", e);
        }
    }

    /**
     * 向RM发送消息
     * @param obj
     */
    public static void sendDtMessageToRm(DtActiveMqBean obj){
        try {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(TOPIC_TO_RM);
            MessageProducer producer = session.createProducer(topic);
            ObjectMessage objectMessage = session.createObjectMessage(obj);
            producer.send(objectMessage);
        } catch (JMSException e) {
            logger.error("向RM发送消息失败", e);
        }
    }

    /**
     * 监听TM消息
     */
    public static void listenerDtMessageForTm(){

        try {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(TOPIC_TO_RM);
            MessageConsumer messageConsumer = session.createConsumer(topic);
            messageConsumer.setMessageListener((message)-> {
                if (message != null && message instanceof ObjectMessage) {
                    ObjectMessage objectMessage = (ObjectMessage) message;
                    try {
                        DtActiveMqBean dtActiveMqBean = (DtActiveMqBean) objectMessage.getObject();
                        logger.info("消费者接收到消息:" + dtActiveMqBean);
                        TransactionClient.informTransaction(dtActiveMqBean);
                    } catch (JMSException e) {
                        logger.error("获取TM消息异常", e);
                    }
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

}
